package io.xxx.mario.task.receive;

import io.xxx.mario.contact.ContactMapper;
import io.xxx.mario.room.RoomMapper;
import io.xxx.mario.task.ReceiverType;
import io.xxx.mario.task.Task;
import jakarta.annotation.Resource;
import org.apache.ibatis.cursor.Cursor;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;

import java.io.IOException;
import java.sql.SQLException;

@Component
public class SystemReceiverSelector implements ReceiverSelector {

    @Resource
    private SqlSessionFactory sqlSessionFactory;

    @Override
    public Flux<Receiver> select(Task task) {
        ReceiverType receiverType = task.getReceiverType();
        String mapperName = receiverType == ReceiverType.ROOM ? RoomMapper.class.getName() : ContactMapper.class.getName();
        return Flux.create(sink -> {
            try (SqlSession sqlSession = sqlSessionFactory.openSession(false)) {
                sqlSession.getConnection().setAutoCommit(false);

                String statement = mapperName + ".findCursorByTaskId";
                try (Cursor<? extends Receiver> cursor = sqlSession.selectCursor(statement, task.getId())) {
                    for (Receiver receiver : cursor) {
                        sink.next(receiver);
                    }
                    sink.complete();
                } catch (IOException e) {
                    sink.error(e);
                }
            } catch (SQLException e) {
                sink.error(e);
            }
        }, FluxSink.OverflowStrategy.BUFFER);
    }
}
